package com.mk.controller;

import com.mk.transaction.SignInService;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Author makai
 * @Description: TODO
 * @Date 2024/1/31 20:15
 */
@RestController
public class MqController {
    @Autowired
    private TransactionMQProducer rocketMQProducer;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private SignInService signInService;
    @Autowired
    private JdbcTemplate jdbcTemplate;

    @GetMapping("/normalMessage")
    public String sendMessage(@RequestParam("msg") String msg) throws Exception {
        SendResult send = rocketMQProducer.send(new Message("test-topic", msg.getBytes()));
        return "Message sent successfully";

    }

    @GetMapping("/delayMessage")
    public String delayMessage(@RequestParam("msg") String msg) throws Exception {
        Message message = new Message("test-delay-topic", ("delayMessage" + msg).getBytes());
        message.setDelayTimeSec(10); //延迟十秒
        rocketMQProducer.send(message);
        return "Message sent successfully";

    }

    @GetMapping("/transactionMessage")
    public String transactionMessage(@RequestParam("msg") String msg) throws Exception {
        Message message = new Message("test-transaction-topic", ("transaction-" + msg).getBytes());
        rocketMQProducer.sendMessageInTransaction(message, 1);
        return "Message sent successfully";
    }


    @GetMapping("/transactionTest")
    public String transactionTest(@RequestParam("no") String no) throws Exception {
        signInService.doSignIn(no);
//        jdbcTemplate.execute("insert into user (id,password,status,user_name) values ('1','1','1','1')");
        return "Message sent successfully";
    }

    @GetMapping("/asyncMessage")
    public String asyncMessage(@RequestParam("msg") String msg) throws Exception {
        Message message = new Message("test-topic", msg.getBytes());
        rocketMQProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步消息发送成功");
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("异步消息发送失败");
            }
        });
        return "Message sent successfully";
    }

    @GetMapping("/orderlyMessage")
    public String orderlyMessage(@RequestParam("msg") String msg) throws Exception {
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload("orderly-" + msg).build();
        //发送顺序消息时，相同类型的消息需要使用相同的hashKey，以此来保证相同类型的消息发送到同一个队列中
        // 顺序消息不能完全100%保证消息消费和生产的顺序行，需要消费者来做最终的保证
        rocketMQTemplate.syncSendOrderly("test-orderly-topic", message, "1");
        return "Message sent successfully";
    }

}
